package org.funtester.performance.books.chapter05.sectioin3;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.net.URI;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Netty WebSocket 客户端
 */
public class NettyWebSocketClient {

    /**
     * Netty 引导程序
     */
    public static Bootstrap bootstrap = new Bootstrap();

    /**
     * EventLoopGroup 线程组
     */
    public static EventLoopGroup group = new NioEventLoopGroup(new ThreadFactory() {

        AtomicInteger index = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("Clinet-" + index.incrementAndGet());
            return thread;
        }
    });

    /**
     * WebSocket 通道组
     */
    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * WebSocket 服务器地址
     */
    public URI uri;

    public WebSocketIOHandler handler;

    /**
     * WebSocket 通道
     */
    public Channel channel;

    static {
        bootstrap.group(group)// 设置 EventLoopGroup
                .channel(NioSocketChannel.class)// 设置 Channel 为 NioSocketChannel
                .option(ChannelOption.TCP_NODELAY, true);// 设置 TCP 为无延迟
    }

    public NettyWebSocketClient(URI serverUri) {
        this.uri = serverUri;
        handler = new WebSocketIOHandler(WebSocketClientHandshakerFactory.newHandshaker(this.uri, WebSocketVersion.V13, null, true, new DefaultHttpHeaders()));// WebSocket 客户端处理器
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {// 设置 ChannelHandler

            @Override
            protected void initChannel(SocketChannel channel) {
                ChannelPipeline pipeline = channel.pipeline();// 获取 Channel 的 ChannelPipeline
                pipeline.addLast(new HttpClientCodec());// 添加 HttpClientCodec, 将请求和应答消息编码或解码为 HTTP 消息
                pipeline.addLast(new HttpObjectAggregator(1024 * 1024));// 添加 HttpObjectAggregator, 将 HTTP 消息的多个部分合成一条完整的 HTTP 消息
                pipeline.addLast(handler);// 添加自定义 WebSocketIOHandler, 处理 WebSocket 消息
            }

        });

    }

    /**
     * 连接 WebSocket 服务器
     * @return
     * @throws InterruptedException
     */
    public ChannelPromise connect() throws InterruptedException {
        this.channel = bootstrap.connect(this.uri.getHost(), this.uri.getPort()).sync().channel();// 连接 WebSocket 服务器
        channels.add(this.channel);// 将连接的 WebSocket 通道添加到 ChannelGroup 中
        return this.handler.handshakeFuture;
    }

    /**
     * 发送 WebSocket 消息
     * @param msg
     * @return
     */
    public ChannelFuture send(String msg) {
        return this.channel.writeAndFlush(new TextWebSocketFrame(msg));// 发送 WebSocket 消息
    }

    /**
     * 关闭 WebSocket 连接
     * @return
     */
    public ChannelFuture close() {
        return this.channel.close();// 关闭 WebSocket 连接
    }

}
